package com.atguigu.flink.exec2;

import com.atguigu.flink.pojo.AdsClickLog;
import com.atguigu.flink.pojo.LoginEvent;
import com.atguigu.flink.utils.MyUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * Created by Smexy on 2023/2/2
 *
 * 如果同一用户（可以是不同IP）在2秒之内连续两次登录，都失败，就认为存在恶意登录的风险
 *
 *      基于个数的窗口： size = 2, slide = 1
 */
public class Demo5_LoginFail
{

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //读数据
        env.readTextFile("data/LoginLog.csv").setParallelism(1)
           .map(new MapFunction<String, LoginEvent>()
           {
               @Override
               public LoginEvent map(String value) throws Exception {
                   String[] words = value.split(",");
                   return new LoginEvent(
                       Long.parseLong(words[0]),
                       words[1],
                       words[2],
                       Long.parseLong(words[3]) * 1000
                   );
               }
           }).setParallelism(1)
           .keyBy(LoginEvent::getUserId)
           .countWindow(2,1)
           .process(new ProcessWindowFunction<LoginEvent, String, Long, GlobalWindow>()
           {
               //elements 最多两条
               @Override
               public void process(Long key, Context context, Iterable<LoginEvent> elements, Collector<String> out) throws Exception {

                   List<LoginEvent> loginEvents = MyUtil.parseList(elements);

                   if (loginEvents.size() == 2){

                       LoginEvent event1 = loginEvents.get(0);
                       LoginEvent event2 = loginEvents.get(1);

                       //在2秒之内连续两次登录，都失败
                       if ("fail".equals(event1.getEventType())
                            &&
                           "fail".equals(event2.getEventType())
                           &&
                           Math.abs(event1.getEventTime() - event2.getEventTime()) <= 2000
                       ){
                           out.collect(key + "在"+ event1.getEventTime() + "," + event2.getEventTime() + "恶意登录... ");
                       }
                   }

               }
           })
           .print().setParallelism(1);

        env.execute();
    }

}
